package com.mall.util.distributed.raft.election;

import com.mall.util.distributed.raft.node.Node;
import com.mall.util.distributed.raft.node.NodeRole;
import com.mall.util.distributed.raft.rpc.DefaultRaftClient;
import com.mall.util.distributed.raft.rpc.RaftClient;
import com.mall.util.distributed.raft.rpc.Request;
import com.mall.util.distributed.raft.rpc.Response;
import com.mall.util.distributed.raft.rpc.message.RequestVote;
import com.mall.util.distributed.raft.rpc.message.ResponseVote;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.net.URI;
import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * 在转变为候选人之后就开始进行选举
 */
@Slf4j
@Getter
@Setter
@RequiredArgsConstructor
public class ElectionTask implements Runnable {
    private final ExecutorService executorService;

    @Autowired
    public RaftClient raftClient;

    @Override
    public void run() {
        var node = Node.getINSTANCE();
        var preElectionTime = node.getPreElectionTimeout();
        var electionTime = node.getElectionTime();

        if (node.isLeader()) return;

        long current = System.currentTimeMillis();

        electionTime = electionTime + ThreadLocalRandom.current().nextInt(50);
        if (current - preElectionTime < electionTime) return;

        /* 将当前结点转换为候选者 */
        node.setNodeRole(NodeRole.CANDIDATE);

        log.warn("node {} will become CANDIDATE and start election leader, current term : [{}], LastEntry : [{}]",
                node.getSelfAddress(), node.getCurrentTerm(), null);

        node.setPreElectionTimeout(System.currentTimeMillis() + ThreadLocalRandom.current().nextInt(200) + 150);

        /* 当前任期增加1 */
        node.getCurrentTerm().incrementAndGet();

        /* 自己投自己 */
        node.setVoteFor(node.getSelfAddress());

        var peers = node.getExcludeSelfPeerAddress();
        var futureArrayList = new ArrayList<Future>();

        /* 发送请求投票消息 */
        for (String peer : peers) {
            futureArrayList.add(executorService.submit(() -> {
                var requestVote = new RequestVote();
                requestVote.setTerm(node.getCurrentTerm().get());
                requestVote.setCandidateId(node.getSelfAddress());
                requestVote.setLastLogIndex(0L);
                requestVote.setLastLogTerm(0L);

                var peerAddress = URI.create(peer);
                var request = Request.newBuilder()
                        .cmd(Request.R_VOTE)
                        .obj(requestVote)
                        .url(peerAddress.getHost() + ":" + peerAddress.getPort())
                        .build();

                try {
                    @SuppressWarnings("unchecked")
                    Response<ResponseVote> response = getRaftClient().send(request);
                    return response;

                } catch (RuntimeException e) {
                    return null;
                }
            }));

            var success = new AtomicInteger(0);
            var countDownLatch = new CountDownLatch(futureArrayList.size());

            for (Future future : futureArrayList) {
                executorService.submit(new Callable<Object>() {
                    @Override
                    public Object call() throws Exception {
                        try {
                            var response = (Response<ResponseVote>) future.get(3000, MILLISECONDS);
                            if (response == null) {
                                return -1;
                            }
                            boolean isVoteGranted = response.getResult().isVoteGranted();
                            if (isVoteGranted) {
                                success.incrementAndGet();
                            } else {
                                // 更新自己的任期.
                                long resTerm = response.getResult().getTerm();
                                if (resTerm >= node.getCurrentTerm().get()) {
                                    node.setCurrentTerm(new AtomicLong(resTerm));
                                }
                            }
                            return 0;
                        }finally {
                            countDownLatch.countDown();
                        }
                    }
                });
            }

            try {
                // 稍等片刻
                countDownLatch.await(3500, MILLISECONDS);
            } catch (InterruptedException e) {
                log.warn("InterruptedException By Master election Task");
            }

            log.info("{} obtain {} ticket", node.getSelfAddress(), success.get());

            if (NodeRole.FOLLOWER.equals(node.getNodeRole())) return;

            if(success.get() > node.getPeerAddress().size() / 2) {
                log.warn("node {} become leader ", node.getSelfAddress());
                node.setNodeRole(NodeRole.LEADER);
                node.resetMark();
                becomeLeaderToDoThing();
            }else {
                node.resetMark();
            }
        }
    }

    /**
     * 初始化所有的index为当前日志模块的索引下标 + 1
     */
    private void becomeLeaderToDoThing() {
        var node = Node.getINSTANCE();
        node.getNextIndexs().clear();
        node.getMatchIndexs().clear();

        for (String peer : node.getExcludeSelfPeerAddress()) {
            node.getMatchIndexs().put(peer,  1L);
            node.getMatchIndexs().put(peer, 0L);
        }
    }
}
